package com.rocketmq.base.example.order;

import com.rocketmq.base.example.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.Random;

/**
 * @Auther: tianchao
 * @Date: 2021/9/2 16:12
 * @Description:
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_order_consumer_group_name");
        consumer.setNamesrvAddr(Const.nameServer);
        consumer.subscribe("OrderTopic", "*");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                Random random = new Random();
                try {
                    Thread.sleep(random.nextInt(10) * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                for (MessageExt msg : msgs) {
                    System.out.println("当前线程"+Thread.currentThread().getName()+"===>"+new String(msg.getBody()));
                }

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
//        consumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//                Random random = new Random();
//                try {
//                    Thread.sleep(random.nextInt(10) * 1000);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//
//                for (MessageExt msg : msgs) {
//                    System.out.println("当前线程"+Thread.currentThread().getName()+"===>"+new String(msg.getBody()));
//                }
//
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
